package com.zw.cdc;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import com.ververica.cdc.connectors.oracle.OracleSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class OracleSourceExample {
  public static void main(String[] args) throws Exception {
	  SourceFunction<String> oracleSource = OracleSource.<String>builder().hostname("192.168.1.222")
			.port(1521)
			.database("orcl")
			.schemaList("AA")
			.tableList("AA.A")
            .username("lzwcdc")
            .password("lzwcdc")
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
    
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    env.setParallelism(1);
//    env.enableCheckpointing(5000);
//    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//    env.getCheckpointConfig().setCheckpointTimeout(1800000L);
//    env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//    env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);
//    env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);
    
    env.addSource(oracleSource).addSink(new PrintSinkFunction<String>());
    
    env.execute("Print ORACLE");
  }
  
  /**
   * oracle 开启补充日志
   */
  // 查询补充日志和强制日志
  // select supplemental_log_data_all,force_logging from v$database;
  
  //alter database add supplemental log data(all) columns;
  //alter database force logging;
  
  /*
   * SQL> select supplemental_log_data_all,force_logging from v$database;

	SUP FORCE_LOGGING
	--- ---------------------------------------
	NO  NO
	
	SQL> alter database add supplemental log data(all) columns;
	
	Database altered.
	
	SQL> alter database force logging;
	
	Database altered.
	
	SQL> select supplemental_log_data_all,force_logging from v$database;
	
	SUP FORCE_LOGGING
	--- ---------------------------------------
	YES YES

   */
  
  /**
   *	 开启oracle 采集用户授权
   *	
   */
 /*
  * 
  *  sqlplus / as sysdba
  *  
	  CREATE TABLESPACE logminer_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/logminer_tbs.dbf'
	    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
	
	  CREATE TABLESPACE logminer_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/ORCLPDB1/logminer_tbs.dbf'
	    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
	
	  CREATE USER lzwcdc IDENTIFIED BY lzwcdc DEFAULT TABLESPACE logminer_tbs QUOTA UNLIMITED ON logminer_tbs;
	
	  GRANT CREATE SESSION TO lzwcdc; 
	  GRANT SET CONTAINER TO lzwcdc; 
	  GRANT SELECT ON V_$DATABASE to lzwcdc; 
	  GRANT FLASHBACK ANY TABLE TO lzwcdc; 
	  GRANT SELECT ANY TABLE TO lzwcdc; 
	  GRANT SELECT_CATALOG_ROLE TO lzwcdc; 
	  GRANT EXECUTE_CATALOG_ROLE TO lzwcdc; 
	  GRANT SELECT ANY TRANSACTION TO lzwcdc; 
	  GRANT LOGMINING TO lzwcdc; 
	
	  GRANT CREATE TABLE TO lzwcdc; 
	  GRANT LOCK ANY TABLE TO lzwcdc; 
	  GRANT CREATE SEQUENCE TO lzwcdc; 
	
	  GRANT EXECUTE ON DBMS_LOGMNR TO lzwcdc; 
	  GRANT EXECUTE ON DBMS_LOGMNR_D TO lzwcdc; 
	
	  GRANT SELECT ON V_$LOG TO lzwcdc; 
	  GRANT SELECT ON V_$LOG_HISTORY TO lzwcdc; 
	  GRANT SELECT ON V_$LOGMNR_LOGS TO lzwcdc; 
	  GRANT SELECT ON V_$LOGMNR_CONTENTS TO lzwcdc; 
	  GRANT SELECT ON V_$LOGMNR_PARAMETERS TO lzwcdc; 
	  GRANT SELECT ON V_$LOGFILE TO lzwcdc; 
	  GRANT SELECT ON V_$ARCHIVED_LOG TO lzwcdc; 
	  GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO lzwcdc; 
	  GRANT SELECT ON V_$TRANSACTION TO lzwcdc; 
	
	  exit;
  */
  
  
  /**
   * GRANT UNLIMITED TABLESPACE TO AA; 
   */
}
